package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo5Burk {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("Demo5Burk")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._
    // 导入Spark SQL所有的函数
    import org.apache.spark.sql.functions._

    // 读取数据
    val burkDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk String,year String,tsl01 String,tsl02 String,tsl03 String,tsl04 String,tsl05 String,tsl06 String,tsl07 String,tsl08 String,tsl09 String,tsl10 String,tsl11 String,tsl12 String")
      .load("spark/data/burks.txt")

    /**
     * 1、统计每个公司每年按月累计收入
     *
     * 输出结果
     * 公司代码,年度,月份,当月收入,累计收入
     */

    // SQL的方式

    burkDF.createOrReplaceTempView("burkT")

    // explode 相当于flatMap
    spark.sql(
      """
        |select burk
        |       ,year
        |       ,month
        |       ,price
        |       ,sum(price) over(partition by burk,year order by month) as sum_price
        |from burkT
        |lateral view explode(Map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) v1 as month,price
        |""".stripMargin)
    //      .show()

    val col: Column = map(
      expr("1"), $"tsl01"
      , expr("2"), $"tsl02"
      , expr("3"), $"tsl03"
      , expr("4"), $"tsl04"
      , expr("5"), $"tsl05"
      , expr("6"), $"tsl06"
      , expr("7"), $"tsl07"
      , expr("8"), $"tsl08"
      , expr("9"), $"tsl09"
      , expr("10"), $"tsl10"
      , expr("11"), $"tsl11"
      , expr("12"), $"tsl12"
    )

    // DSL
    burkDF
      .select($"burk", $"year", explode(col) as Array("month", "price"))
      .select($"burk", $"year", $"month", $"price", sum($"price").over(Window.partitionBy($"burk", $"year").orderBy($"month")) as "sum_price")
    //      .show()

    /**
     * 2、统计每个公司当月比上年同期增长率
     * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
     */

    burkDF
      .select($"burk", $"year", explode(col) as Array("month", "price"))
      .select($"burk", $"year", $"month", $"price", lag($"price", 1).over(Window.partitionBy($"burk", $"month").orderBy($"year")) as "last_price")
      // coalesce(v1,v2,v3,v4) 从左往右依次取第一个非空的值
      .select($"burk", $"year", $"month", round(coalesce($"price" / $"last_price" - 1, expr("1")), 5) as "inc")
      .show()
  }

}
